package com.fwmagic.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

public class HelloWorldObserver extends BaseRegionObserver {

    public static final String JACK = "JACK";

    Logger logger = LoggerFactory.getLogger(HelloWorldObserver.class);


    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {

        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "hd1:2181,hd2:2181,hd3:2181");
        Connection connection = ConnectionFactory.createConnection(config);

        Table table = connection.getTable(TableName.valueOf("mytable"));
        //获取mycf:name单元格
        List<Cell> name = put.get(Bytes.toBytes("mycf"), Bytes.toBytes("name"));
        logger.error("======> name:"+name.size()+"===="+name);
        //如果该PUT中不存在mycf:name这个单元格则直接返回
        if (name == null || name.size() == 0) {
            return;
        }

        //比较mycf:name是否为JACK
        byte[] bytes = CellUtil.cloneValue(name.get(0));
        String s = Bytes.toString(bytes);
        logger.error("=======>s:"+s);
        if (JACK.equals(s)) {
            //在mycf:message中添加一句话
            put.addColumn(Bytes.toBytes("mycf"), Bytes.toBytes("message"), Bytes.toBytes("Hello world!Welcome back!"));
        }
        table.put(put);

        table.close();
        connection.close();
    }
}
